/* NOTICE
    OpenJaWS - Open Java Weather Station
    
    Copyright (C) 2008 Grant Gardner <grant@lastweekend.com.au>
    
    OpenJaWS is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
    
NOTICE */

package au.com.lastweekend.openjaws.plugins;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import au.com.lastweekend.openjaws.WeatherLoop;
import au.com.lastweekend.openjaws.api.Readings;
import au.com.lastweekend.openjaws.api.WeatherPlugin;
import au.com.lastweekend.openjaws.api.WeatherPluginException;
import au.com.lastweekend.openjaws.util.LoopRunnable;

/**
 * Clones readings and hands off to another thread.
 * 
 * Useful for plugins that may occasionally run for a long time (eg website updates) and would otherwise slow down all readings.
 * 
 * If the queue is more than {@link #getQueueCapacity()} readings behind then the thread reading this process (typically the main
 * {@link WeatherLoop} will be blocked.
 * 
 * @author grant@lastweekend.com.au
 * @version $Id: codetemplates.xml,v 1.2 2005/01/22 06:53:11 ggardner Exp $
 */
public class QueuedReadingsPlugin implements WeatherPlugin {

    public static final Logger LOG = LoggerFactory.getLogger(QueuedReadingsPlugin.class);

    public class ReadingsConsumer extends LoopRunnable {

        @Override
        protected void inLoop() {

            try {
                Readings readings = queue.take();
                delegateProcess(readings);
            } catch (InterruptedException e) {
                LOG.debug("Interrupted waiting on queue size=" + queue.size(), e);
            }

        }

        @Override
        protected void stopped() {

            List<Readings> sink = new LinkedList<Readings>();
            queue.drainTo(sink);
            if (LOG.isDebugEnabled()) {
                LOG.debug("Drained " + sink.size() + " readings from waiting queue");
            }
            for (Readings readings : sink) {
                delegateProcess(readings);
            }
            LOG.info("Stopped");
        }

        private synchronized void delegateProcess(Readings taken) {

            Thread.interrupted();
            try {
                if (LOG.isDebugEnabled()) {
                    LOG.debug(String.format("Processing queued reading %tc", taken.getReadingTime()));
                }
                delegate.process(taken);
            } catch (WeatherPluginException e) {
                LOG.error("Exception processing queued reading", e);
            }

        }

    }

    private int queueCapacity = 100;
    private WeatherPlugin delegate;
    private BlockingQueue<Readings> queue;
    private ReadingsConsumer readingsConsumer;

    public QueuedReadingsPlugin(WeatherPlugin delegate) {

        this.delegate = delegate;

    }

    @PostConstruct
    public void start() {

        if (queue == null) {
            queue = new ArrayBlockingQueue<Readings>(queueCapacity);
            readingsConsumer = new ReadingsConsumer();
            readingsConsumer.start("ReadingsQueue");
        }

    }

    @PreDestroy
    public void stop() {

        if (readingsConsumer != null) {
            readingsConsumer.stop();
        }

    }

    public void stopWait(long timeout) throws InterruptedException {

        if (readingsConsumer != null) {
            readingsConsumer.stopWait(timeout);
        }
    }

    public void process(Readings readings) throws WeatherPluginException {

        if (queue == null) {
            start();
        }

        Readings clone = new Readings(readings);

        try {
            queue.put(clone);
            if (LOG.isDebugEnabled()) {
                LOG.debug("Queue size=" + queue.size());
            }
        } catch (InterruptedException e) {
            throw new WeatherPluginException("Interrupted processing reading", e);
        }

    }

    public int getQueueCapacity() {

        return queueCapacity;
    }

    public void setQueueCapacity(int queueCapacity) {

        this.queueCapacity = queueCapacity;
    }

}
